package flink.demo;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SideOutput_Demo {
    public static void main(String[] args) throws Exception {
        Configuration configuration=new Configuration();
        configuration.setInteger("rest.port",8822);



        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);



        // 开启checkpoint
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
       env.getCheckpointConfig().setCheckpointStorage("file:///D:\\data\\chk");


        DataStreamSource<MySourceFunction.EventLog> streamSource = env.addSource(new MySourceFunction());
        // 需求： 将行为事件流，进行分流
        //     appLaunch 事件 ，分到一个流
        //     putBack 事件，分到一个流
        //     其他事件保留在主流

        SingleOutputStreamOperator<MySourceFunction.EventLog> process = streamSource.process(new ProcessFunction<MySourceFunction.EventLog, MySourceFunction.EventLog>() {
            @Override
            public void processElement(MySourceFunction.EventLog eventLog, Context context, Collector<MySourceFunction.EventLog> out) throws Exception {
                if ("appLaunch".equals(eventLog.getEventId())) {
                    context.output(new OutputTag<MySourceFunction.EventLog>("launch", TypeInformation.of(MySourceFunction.EventLog.class)), eventLog);
                } else if ("putBack".equals(eventLog.getEventId())) {
                    context.output(new OutputTag<MySourceFunction.EventLog>("back",TypeInformation.of(MySourceFunction.EventLog.class)), eventLog);

                }
                out.collect(eventLog);
            }
        });

        DataStream<MySourceFunction.EventLog> launch = process.getSideOutput(new OutputTag<MySourceFunction.EventLog>("launch",TypeInformation.of(MySourceFunction.EventLog.class)));
        DataStream<MySourceFunction.EventLog> back = process.getSideOutput(new OutputTag<MySourceFunction.EventLog>("back",TypeInformation.of(MySourceFunction.EventLog.class)));

        launch.print("launch");
        back.print("back");
        env.execute();

    }


}
